package zjs.dc.controller;

import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;

/**
 * @Class:
 * @Author: guoml
 * @Description: 针对企查查t_ipo_company_senior_executive表和聚源LC_LeaderIntroduce表进行数据映射开发
 * @Date: 2021/12/7 12:51
 */

public class TIpoCompanySeniorExecutive {
    public static void main(String[] args) {
        SparkSession spark = SparkSession.builder().appName("t_ipo_company_senior_executive").enableHiveSupport().getOrCreate();
        spark.sparkContext().setLogLevel("ERROR");
        //获取中数增量数据表中的新增数据（left join）
        Dataset<Row> insert_data = spark.sql("SELECT md5(concat(100,aaa.id)),\n" +
                "       main.SecuCode,\n" +
                "       aaa.leadername,\n" +
                "       aaa.gender,\n" +
                "       aaa.age,\n" +
                "       aaa.degree,\n" +
                "       aaa.position,\n" +
                "       '' as created_date,\n" +
                "       '' as updated_date,\n" +
                "       '' as encrypt_key,\n" +
                "       regexp_replace(current_date(),'-',''),\n" +
                "       \"1\" as isadd\n" +
                "FROM\n" +
                "  (SELECT aa.id,\n" +
                "          aa.companycode,\n" +
                "          aa.leadername,\n" +
                "          if(aa.LeaderGender=1,'男','女') AS gender,\n" +
                "          floor((UNIX_TIMESTAMP(CURRENT_DATE) - UNIX_TIMESTAMP(substr(aa.BirthYM,0,10),'yyyyMMdd'))/(365*24*60*60)) AS age,\n" +
                "          CASE aa.leaderDegree\n" +
                "              WHEN 1 THEN '博士后'\n" +
                "              WHEN 2 THEN '博士'\n" +
                "              WHEN 3 THEN '硕士'\n" +
                "              WHEN 4 THEN '本科'\n" +
                "              WHEN 5 THEN '大专'\n" +
                "              WHEN 6 THEN '高中'\n" +
                "              WHEN 7 THEN '中专'\n" +
                "              ELSE '其他'\n" +
                "          END AS degree,\n" +
                "          aa.InfoPublDate,\n" +
                "          aa.XGRQ,\n" +
                "          replace(concat_ws(\"\\\\\",collect_set(aa.ms)),'\\\\','，') as position\n" +
                "   FROM\n" +
                "     (SELECT a.id,\n" +
                "             a.companycode,\n" +
                "             a.LeaderGender,\n" +
                "             a.BirthYM,\n" +
                "             a.BirthYMInfo,\n" +
                "             a.LeaderDegree,\n" +
                "             a.InfoPublDate,\n" +
                "             a.XGRQ,\n" +
                "             a.leadername,\n" +
                "             c.ms\n" +
                "      FROM\n" +
                "        (SELECT *\n" +
                "         FROM a_dc_ep_ods.LC_LeaderIntroduce\n" +
                "         WHERE IfIn = '1') a\n" +
                "      LEFT JOIN a_dc_ep_ods.LC_LeaderIntroduce_SE b ON a.id = b.id\n" +
                "      AND b.TypeCode = 1\n" +
                "      LEFT JOIN\n" +
                "        (SELECT *\n" +
                "         FROM a_dc_ep_ods.CT_SystemConst\n" +
                "         WHERE LB ='1182') c ON b.Code = c.DM) aa\n" +
                "   GROUP BY aa.id,\n" +
                "            aa.companycode,\n" +
                "            aa.leadername,\n" +
                "            aa.LeaderGender,\n" +
                "            aa.BirthYM,\n" +
                "            aa.BirthYMInfo,\n" +
                "            aa.LeaderDegree,\n" +
                "            aa.InfoPublDate,\n" +
                "            aa.XGRQ) aaa\n" +
                "INNER JOIN\n" +
                "  (SELECT *\n" +
                "   FROM a_dc_ep_ods.secumain\n" +
                "   WHERE SecuCategory=1\n" +
                "     AND ListedState=1 AND ListedSector not in (3,7)) main ON aaa.companycode = main.companycode").union(spark.sql("SELECT md5(concat(200,aaa.id)),\n" +
                "       main.SecuCode,\n" +
                "       aaa.leadername,\n" +
                "       aaa.gender,\n" +
                "       aaa.age,\n" +
                "       aaa.EducationLevel,\n" +
                "       aaa.position,\n" +
                "       '' as created_date,\n" +
                "       '' as updated_date,\n" +
                "       '' as encrypt_key,\n" +
                "       regexp_replace(current_date(),'-',''),\n" +
                "       \"1\" as isadd\n" +
                "FROM\n" +
                "  (SELECT aa.id,\n" +
                "          aa.companycode,\n" +
                "          aa.leadername,\n" +
                "          aa.Gender,\n" +
                "          aa.age AS age,\n" +
                "          aa.EducationLevel,\n" +
                "          replace(concat_ws(\"\\\\\",collect_set(aa.PosnName)),'\\\\','，') as position\n" +
                "   FROM\n" +
                "     (SELECT a.id,\n" +
                "             a.companycode,\n" +
                "             a.Gender,\n" +
                "             b.age,\n" +
                "             a.EducationLevel,\n" +
                "             a.PosnName,\n" +
                "             a.leadername\n" +
                "      FROM\n" +
                "        (SELECT *\n" +
                "         FROM a_dc_ep_ods.NQ_LeaderPOSN\n" +
                "         WHERE IfPosition = '1') a\n" +
                "      LEFT JOIN a_dc_ep_ods.NQ_PersonalInfo b ON a.PersonalCode = b.PersonalNum) aa\n" +
                "   GROUP BY aa.id,\n" +
                "            aa.companycode,\n" +
                "            aa.leadername,\n" +
                "            aa.Gender,\n" +
                "            aa.age,\n" +
                "            aa.EducationLevel) aaa\n" +
                "INNER JOIN\n" +
                "  (SELECT *\n" +
                "   FROM a_dc_ep_ods.nq_secumain\n" +
                "   WHERE SecuCategory=1\n" +
                "     AND ListedState=1 and ListedSector =3) main ON aaa.companycode = main.companycode")).union(spark.sql("SELECT md5(concat(300,aaa.id)),\n" +
                "       main.SecuCode,\n" +
                "       aaa.leadername,\n" +
                "       aaa.gender,\n" +
                "       aaa.age,\n" +
                "       aaa.degree,\n" +
                "       aaa.position,\n" +
                "       '' as created_date,\n" +
                "       '' as updated_date,\n" +
                "       '' as encrypt_key,\n" +
                "       regexp_replace(current_date(),'-',''),\n" +
                "       \"1\" as isadd\n" +
                "FROM\n" +
                "  (SELECT aa.id,\n" +
                "          aa.companycode,\n" +
                "          aa.leadername,\n" +
                "          if(aa.Gender=1,'男','女') AS gender,\n" +
                "          floor((UNIX_TIMESTAMP(CURRENT_DATE) - UNIX_TIMESTAMP(substr(aa.BirthYM,0,10),'yyyyMMdd'))/(365*24*60*60)) AS age,\n" +
                "          CASE aa.EducationLevel\n" +
                "              WHEN 1 THEN '博士后'\n" +
                "              WHEN 2 THEN '博士'\n" +
                "              WHEN 3 THEN '硕士'\n" +
                "              WHEN 4 THEN '本科'\n" +
                "              WHEN 5 THEN '大专'\n" +
                "              WHEN 6 THEN '高中'\n" +
                "              WHEN 7 THEN '中专'\n" +
                "              ELSE '其他'\n" +
                "          END AS degree,\n" +
                "          replace(concat_ws(\"\\\\\",collect_set(aa.ms)),'\\\\','，') as position\n" +
                "   FROM\n" +
                "     (SELECT a.id,\n" +
                "             a.companycode,\n" +
                "             a.Gender,\n" +
                "             a.BirthYM,\n" +
                "             a.EducationLevel,\n" +
                "             a.leadername,\n" +
                "             c.ms\n" +
                "      FROM\n" +
                "        (SELECT *\n" +
                "         FROM a_dc_ep_ods.LC_STIBLeaderIntroduce\n" +
                "         WHERE Incumbent = '1') a\n" +
                "      LEFT JOIN a_dc_ep_ods.LC_STIBLeaderIntroduce_SE b ON a.id = b.id\n" +
                "      AND b.TypeCode = 1\n" +
                "      LEFT JOIN\n" +
                "        (SELECT *\n" +
                "         FROM a_dc_ep_ods.CT_SystemConst\n" +
                "         WHERE LB ='1182') c ON b.Code = c.DM) aa\n" +
                "   GROUP BY aa.id,\n" +
                "            aa.companycode,\n" +
                "            aa.leadername,\n" +
                "            aa.Gender,\n" +
                "            aa.BirthYM,\n" +
                "            aa.EducationLevel) aaa\n" +
                "INNER JOIN\n" +
                "  (SELECT *\n" +
                "   FROM a_dc_ep_ods.secumain\n" +
                "   WHERE SecuCategory=1\n" +
                "     AND ListedState=1 and ListedSector =7) main ON aaa.companycode = main.companycode"));
        insert_data.createOrReplaceTempView("tmp_t_ipo_company_senior_executive1");
        spark.sql("insert overwrite table a_dc_ep_incr.t_ipo_company_senior_executive select * from tmp_t_ipo_company_senior_executive1");
        spark.stop();
    }
}
